﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

using Microsoft.Practices.ServiceLocation;

#region WorkStation Framework Using

using WorkStation.Frameworks.Core;
using WorkStation.Frameworks.Core.Extensions;
using WorkStation.Frameworks.EasyBus.Handles;
using WorkStation.Frameworks.EasyBus.MessageQueue;
using WorkStation.Frameworks.EasyBus.Pipeline;
using WorkStation.Frameworks.EasyBus.Pipeline.Modules;
using WorkStation.Frameworks.EasyBus.Pipeline.Extensions;
using WorkStation.Frameworks.EasyBus.Pipeline.Handles;

#endregion

namespace WorkStation.Frameworks.EasyBus
{
    /// <summary>
    /// 消息监听器
    /// </summary>
    public class MessageListener : IMessageListener
    {
        #region 事件

        public event Action<BusActionContext, FailHandleResult> OnProcessFail;
        public event Action<BusActionContext, SuccessHandleResult> OnProcessSuccess;
        public event Action<BusActionContext> OnProcessBefore;
        public event Action<BusActionContext, IEnumerable<HandleResult>> OnProcessAfter;

        public event Action<Exception> OnListenException;

        #endregion

        #region 私有变量

        protected List<Task> _tasks = new List<Task>();
        protected CancellationToken _token;
        protected CancellationTokenSource _source;
        protected bool _isAlreadyInitPipelineEvent = false;
        protected IChannel _channel = null;

        #endregion

        #region 公有属性

        public string Name { get; set; }

        public IEndpoint Endpoint { get; set; }

        public IPipeline ReceiveMessagePipeline { get; set; }

        public IBus Bus { get; set; }

        public int ProcessThreadCount { get; set; }

        public bool Enabled { get; set; }

        #endregion

        public MessageListener(IBus bus, IEndpoint endpoint) 
        {
            this.ReceiveMessagePipeline = MethodPipeline<IPipeline>.Wrap(() => ServiceLocator.Current.GetInstance<IPipeline>())
                                                                   .Error((args, res) => res.Result = new ObjectPipeline())
                                                                   .Execute()
                                                                   .Result;

            if (bus == null) 
            {
                throw new ArgumentNullException("bus");
            }

            if (endpoint == null) 
            {
                throw new ArgumentNullException("endpoint");
            }

            this.Bus = bus;
            this.Endpoint = endpoint;
            this.ProcessThreadCount = 1;
            this.Enabled = true;

            this.ReceiveMessagePipeline.Subscribe<object>(new ProcessRouteHandle() { Listener = this });
        }

        public IMessageListener Subscribe(string msgType, IHandle handle)
        {
            this.ReceiveMessagePipeline.Subscribe(msgType, handle);

            return this;
        }

        public void BeginListen()
        {
            this.InitReceiveMessagePipelineEvent();

            if (!this.Enabled) { return; }

            this._source = new CancellationTokenSource();

            this._token = this._source.Token;

            this._channel = this.Endpoint.GetChannel();

            for (int i = 0; i < this.ProcessThreadCount; i++) 
            {
                this._tasks.Add(this.CreateThread());
            }

            this._tasks.ForEach(v => v.Start());
        }

        public void StopListen()
        {
            this._source.Cancel();

            // 取消时调用 Task.WaitAll 会抛出未处理的异常（此处忽略处理取消异常）
            try
            {
                Task.WaitAll(this._tasks.ToArray());
            }
            catch (AggregateException ce) 
            {
                ce.Handle(exception => true);
            }

            if (this._channel != null) 
            {
                this._channel.Dispose();
            }

            this._tasks.Clear();
        }

        protected virtual Task CreateThread() 
        {
            Task task = new Task(() =>
            {
                while (true)
                {                    
                    if (this._token.IsCancellationRequested)
                    {
                        this._token.ThrowIfCancellationRequested();
                    }

                    Message msg = null;

                    try
                    {
                        msg = this._channel.ReadMsg();
                    }
                    catch (Exception ce)
                    {
                        this.Bus.Logger.Error("Listener read message error \r\n{0}".Fmt(ce.ToString()));

                        if (this.OnListenException != null)
                        {
                            this.OnListenException(ce);
                        }
                    }

                    if (msg == null) { continue; }

                    this.Bus.Logger.Info("Process message, message body: \r\n{0}".Fmt(msg.BodyJson));

                    if (this.ReceiveMessagePipeline != null)
                    {
                        if (this.Endpoint.IsSynchronization)
                        {
                            this.ReceiveMessagePipeline.Publish(new BusActionContext() { Bus = this.Bus, Input = msg, RouteKey = msg.RouteKey });
                        }
                        else
                        {
                            // 使用 TaskCreationOptions.AttachedToParent 时，父任务结束时等待子任务执行完后再结束，此时父任务的状态为 TaskStatus.WaitingForChildrenToComplete，此时就算调用取消也会等子任务执行完。
                            Task.Factory.StartNew((m) =>
                            {
                                Message message = m as Message;
                                this.ReceiveMessagePipeline.Publish(new BusActionContext() { Bus = this.Bus, Input = message, RouteKey = message.RouteKey });
                            }, msg, TaskCreationOptions.AttachedToParent);
                        }
                    }
                }
            }, this._token);
            
            task.ContinueWith(v =>
            {
                if (v.IsFaulted)
                {
                    // 调用 AggregateException.Hanlde 处理异常，返回 true 不抛出，否则继续上抛
                    v.Exception.Handle(exception => 
                    {
                        if (exception is OperationCanceledException)
                        {
                            return false;
                        }
                        else if (this.OnListenException != null)
                        {
                            this.OnListenException(exception);
                            return true;
                        }
                        else 
                        {
                            return false;
                        }
                    });
                }
            });

            return task;
        }

        protected virtual void InitReceiveMessagePipelineEvent()
        {
            if (this._isAlreadyInitPipelineEvent) { return; }

            if (this.OnProcessFail != null)
            {
                this.ReceiveMessagePipeline.OnFail += (context, result) => this.OnProcessFail(context as BusActionContext, result);
            }

            if (this.OnProcessSuccess != null)
            {
                this.ReceiveMessagePipeline.OnSuccess += (context, result) => this.OnProcessSuccess(context as BusActionContext, result);
            }

            if (this.OnProcessBefore != null)
            {
                this.ReceiveMessagePipeline.OnPublishBefore += context => this.OnProcessBefore(context as BusActionContext);
            }

            if (this.OnProcessAfter != null)
            {
                this.ReceiveMessagePipeline.OnPublishAfter += (context, result) => this.OnProcessAfter(context as BusActionContext, result);
            }

            this._isAlreadyInitPipelineEvent = true;
        }
    }
}
